package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.data_etl.pojos.SearchAggInfo;
import cn.doitedu.rtdw.data_etl.pojos.SearchAggRes;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;

import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

public class Job5_异步_搜索行为olap分析主题轻度聚合模型 {
    public static void main(String[] args) throws Exception {
        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        //env.setStateBackend(new HashMapStateBackend());
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));



        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * 创建逻辑表（连接器表），映射kafka中物理topic（维度打宽明细行为日志）
         */
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                          "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_Id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     url_prefix    STRING,                         "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     proc_time AS proctime(),                      " // processing time 时间语义
                        + "     rt AS  to_timestamp_ltz(event_time,3),        " // 表达式字段，用于将event_time转成timestamp(3)类型
                        + "     WATERMARK FOR rt AS  rt - INTERVAL '0' SECOND " // 基于rt字段定义watermark，从此，rt字段就具备了 flink中事件时间语义
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd_events',                          "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           "
        );


        /**
         *   搜索行为事实数据提取
         */
        tenv.executeSql(
                " CREATE TEMPORARY VIEW  search_events AS                     " +
                        " SELECT                                                       " +
                        "   rt,                                                        " +
                        "   user_id,                                                   " +
                        " 	event_Id,                                                  " +
                        " 	event_time,                                                " +
                        " 	properties['search_id'] as search_id,                      " +
                        " 	properties['keyword'] as keyword,                          " +
                        " 	cast(properties['res_cnt'] as bigint) as res_cnt           " +
                        " FROM dwd_kafka                                               " +
                        " WHERE event_Id in ('search','search_return','search_click')  "
        );

        //tenv.executeSql("select * from search_events").print();
        /*
+----+---------+----------------+----------------------+------------+----------------+----------+
| op | user_id |       event_Id |           event_time |  search_id |        keyword |  res_cnt |
+----+---------+----------------+----------------------+------------+----------------+----------+
| +I |       3 |         search |        1670596213000 |       sc01 |   usb 移动固态 |     <NULL> |
| +I |       3 |  search_return |        1670596214000 |       sc01 |   usb 移动固态 |        276 |

| +I |       3 |   search_click |        1670596216000 |       sc01 |   usb 移动固态 |     <NULL> |
| +I |       3 |   search_click |        1670596224000 |       sc01 |   usb 移动固态 |     <NULL> |

| +I |       3 |         search |        1670596215000 |       sc02 |   固态移动硬盘  |      NULL> |
| +I |       3 |  search_return |        1670596216000 |       sc02 |   固态移动硬盘  |        276 |

| +I |       5 |         search |        1670596215000 |       sc03 |     速溶苦咖啡 |     <NULL> |
| +I |       5 |  search_return |        1670596216000 |       sc03 |     速溶苦咖啡 |        186 |
| +I |       5 |   search_click |        1670596217000 |       sc03 |     速溶苦咖啡 |     <NULL> |
| +I |       5 |   search_click |        1670596222000 |       sc03 |     速溶苦咖啡 |     <NULL> |
| +I |       5 |   search_click |        1670596224000 |       sc03 |     速溶苦咖啡 |     <NULL> |
        */

        tenv.executeSql(
                " CREATE TEMPORARY VIEW  agg_per_search AS                            " +
                        " SELECT                                                              " +
                        "     window_start,                                                   " +
                        " 	window_end,                                                       " +
                        " 	user_id,                                                          " +
                        " 	search_id,                                                        " +
                        " 	keyword,                                                          " +
                        " 	min(event_time) as search_time,                                   " +
                        " 	max(res_cnt) as res_cnt,                                          " +
                        " 	count(1) FILTER(WHERE event_Id = 'search_click') as click_cnt     " +
                        " FROM TABLE(                                                         " +
                        "     TUMBLE(TABLE search_events,DESCRIPTOR(rt),INTERVAL '1' MINUTE)  " +
                        " )                                                                   " +
                        " GROUP BY                                                            " +
                        "     window_start,                                                   " +
                        " 	window_end,                                                       " +
                        " 	user_id,                                                          " +
                        " 	search_id,                                                        " +
                        " 	keyword                                                           "
        );

        //tenv.executeSql("select * from agg_per_search").print();
/*
+----+-------------------------+-------------------------+----------+------------+---------------+----------------------+--------------+-------------+
| op |            window_start |              window_end |  user_id |  search_id |       keyword |          search_time |      res_cnt |   click_cnt |
+----+-------------------------+-------------------------+----------+------------+---------------+----------------------+--------------+-------------+
| +I | 2022-12-09 22:30:00.000 | 2022-12-09 22:31:00.000 |        3 |       sc01 |    usb 移动固态 |        1670596213000 |          276 |           2 |
| +I | 2022-12-09 22:30:00.000 | 2022-12-09 22:31:00.000 |        5 |       sc03 |      速溶苦咖啡 |        1670596215000 |          186 |           3 |
| +I | 2022-12-09 22:30:00.000 | 2022-12-09 22:31:00.000 |        3 |       sc02 |    固态移动硬盘 |        1670596215000 |          276 |           0 |
 */


        /* *
         * 关联 搜索词分词  和  搜索词标准近义词
         */
        Table table = tenv.from("agg_per_search");
        DataStream<SearchAggInfo> dataStream = tenv.toDataStream(table, SearchAggInfo.class);
        // 小细节优化
        // 此处keyBy（keyword），那么未来相同搜索词会发送到相同下游subTask，这样在subTask中对请求过的词缓存结果，有利于提升效率

        KeyedStream<SearchAggInfo, String> keyedStream = dataStream.keyBy(SearchAggInfo::getKeyword);
        SingleOutputStreamOperator<SearchAggRes> resultStream1 =keyedStream
                        .process(new KeyedProcessFunction<String, SearchAggInfo, SearchAggRes>() {

                            CloseableHttpClient client;
                            HttpPost post;
                            JSONObject jsonObject;
                            SearchAggRes searchAggRes;


                            @Override
                            public void open(Configuration parameters) throws Exception {

                                client = HttpClientBuilder.create().build();

                                post = new HttpPost("http://192.168.77.88:8081/api/post/simwords");
                                post.addHeader("Content-type", "application/json; charset=utf-8");
                                post.addHeader("Accept", "application/json");

                                jsonObject = new JSONObject();

                                searchAggRes = new SearchAggRes();

                            }

                            @Override
                            public void processElement(SearchAggInfo searchAggInfo, KeyedProcessFunction<String, SearchAggInfo, SearchAggRes>.Context ctx, Collector<SearchAggRes> out) throws Exception {

                                // 获取输入数据中的 搜索词
                                String keyword = searchAggInfo.getKeyword();

                                // 将搜索词，放入post的请求体参数中
                                jsonObject.put("origin", keyword);
                                post.setEntity(new StringEntity(jsonObject.toJSONString(), "utf-8"));

                                // 发起请求
                                CloseableHttpResponse response = client.execute(post);
                                // 获取响应体
                                HttpEntity entity = response.getEntity();
                                // 将响应体转成字符串
                                String responseJson = EntityUtils.toString(entity);


                        /*
                               {
                                  "origin":"usb 移动固态",
                                  "words":"usb|移动|固态|",
                                  "similarWord":"移动固态硬盘"
                               }
                         */
                                // 解析结果
                                JSONObject resObject = JSON.parseObject(responseJson);
                                String split_words = resObject.getString("words");
                                String similar_word = resObject.getString("similarWord");


                                // 返回最终结果：输入的数据  +  分词 + 近义词
                                searchAggRes.setWindow_start(searchAggInfo.getWindow_start().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                                searchAggRes.setWindow_end(searchAggInfo.getWindow_end().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                                searchAggRes.setUser_id(searchAggInfo.getUser_id());
                                searchAggRes.setSearch_id(searchAggInfo.getSearch_id());
                                searchAggRes.setKeyword(searchAggInfo.getKeyword());
                                searchAggRes.setSearch_time(searchAggInfo.getSearch_time());
                                searchAggRes.setRes_cnt(searchAggInfo.getRes_cnt());
                                searchAggRes.setClick_cnt(searchAggInfo.getClick_cnt());
                                searchAggRes.setSplit_words(split_words);
                                searchAggRes.setSimilar_word(similar_word);

                                out.collect(searchAggRes);

                            }
                        });


        SingleOutputStreamOperator<SearchAggRes> resultStream =
                AsyncDataStream.orderedWait(keyedStream, new RichAsyncFunction<SearchAggInfo, SearchAggRes>() {
            CloseableHttpClient client;
            HttpPost post;
            JSONObject jsonObject;
            SearchAggRes searchAggRes;

            @Override
            public void open(Configuration parameters) throws Exception {

                client = HttpClientBuilder.create().build();
                post = new HttpPost("http://192.168.77.88:8081/api/post/simwords");
                post.addHeader("Content-type", "application/json; charset=utf-8");
                post.addHeader("Accept", "application/json");
                jsonObject = new JSONObject();
                searchAggRes = new SearchAggRes();

            }

            @Override
            public void asyncInvoke(SearchAggInfo searchAggInfo, ResultFuture<SearchAggRes> resultFuture) throws Exception {

                // 获取输入数据中的 搜索词
                String keyword = searchAggInfo.getKeyword();

                // 将搜索词，放入post的请求体参数中
                jsonObject.put("origin", keyword);
                post.setEntity(new StringEntity(jsonObject.toJSONString(), "utf-8"));

                // 发起请求
                CloseableHttpResponse response = client.execute(post);
                // 获取响应体
                HttpEntity entity = response.getEntity();
                // 将响应体转成字符串
                String responseJson = EntityUtils.toString(entity);


                        /*
                               {
                                  "origin":"usb 移动固态",
                                  "words":"usb|移动|固态|",
                                  "similarWord":"移动固态硬盘"
                               }
                         */
                // 解析结果
                JSONObject resObject = JSON.parseObject(responseJson);
                String split_words = resObject.getString("words");
                String similar_word = resObject.getString("similarWord");


                // 返回最终结果：输入的数据  +  分词 + 近义词
                searchAggRes.setWindow_start(searchAggInfo.getWindow_start().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                searchAggRes.setWindow_end(searchAggInfo.getWindow_end().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                searchAggRes.setUser_id(searchAggInfo.getUser_id());
                searchAggRes.setSearch_id(searchAggInfo.getSearch_id());
                searchAggRes.setKeyword(searchAggInfo.getKeyword());
                searchAggRes.setSearch_time(searchAggInfo.getSearch_time());
                searchAggRes.setRes_cnt(searchAggInfo.getRes_cnt());
                searchAggRes.setClick_cnt(searchAggInfo.getClick_cnt());
                searchAggRes.setSplit_words(split_words);
                searchAggRes.setSimilar_word(similar_word);


                // 返回结果
                resultFuture.complete(Collections.singletonList(searchAggRes));



            }
        }, 1000, TimeUnit.MICROSECONDS);



        // 将结果流转成视图
        tenv.createTemporaryView("res", resultStream);


        // 打印结果
        tenv.executeSql("select * from res").print();

        env.execute();


    }

}
